package com.ruoyi.web.controller.webSocket;

import java.io.IOException;
import java.net.URLDecoder;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Semaphore;
import javax.websocket.OnClose;
import javax.websocket.OnError;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.ServerEndpoint;

import cn.hutool.core.collection.ConcurrentHashSet;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.ruoyi.common.constant.ReceivedMessage;
import com.ruoyi.common.core.redis.RedisCache;
import com.ruoyi.common.utils.DeepLTranslationUtil;
import com.ruoyi.common.utils.StringUtils;
import com.ruoyi.common.utils.spring.SpringUtils;
import com.ruoyi.framework.websocket.MessageEncoder;
import com.ruoyi.framework.websocket.SemaphoreUtils;
import com.ruoyi.framework.websocket.WebSocketConfig;
import com.ruoyi.framework.websocket.WebSocketUsers;
import com.ruoyi.smartai.utils.WhatsAPPUtils;
import com.ruoyi.system.domain.ChatLog;
import com.ruoyi.system.mapper.ChatLogMapper;
import com.ruoyi.system.service.iChatLogService;
import com.ruoyi.system.service.impl.WaURLService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

/**
 * websocket 消息处理
 *
 * @author ruoyi
 */
@Component
@ServerEndpoint(value = "/websocket/message", encoders = {MessageEncoder.class})
public class WebSocketServer {

    @Autowired
    RedisCache redisCache;

    @Autowired
    private ChatLogMapper chatLogMapper;

    /**
     * WebSocketServer 日志控制器
     */
    private static final Logger LOGGER = LoggerFactory.getLogger(WebSocketServer.class);

    // 使用静态成员来保存RedisCache实例
    private static final ConcurrentHashMap<String, Session> sessionMap = new ConcurrentHashMap<>();

    private static final ObjectMapper objectMapper = new ObjectMapper();

    /**
     * 默认最多允许同时在线人数100
     */
    public static int socketMaxOnlineCount = 2000;

    public static Semaphore socketSemaphore = new Semaphore(socketMaxOnlineCount);

    // 单例实例
    private static volatile WebSocketServer instance;

    private static WaURLService waURLService;
    private static Map<String, Set<String>> numberToSessionKeys = new ConcurrentHashMap<>();


    static {
        waURLService = SpringUtils.getBean(WaURLService.class);
    }

    /**
     * 连接建立成功调用的方法
     */
    @OnOpen
    public void onOpen(Session session) throws Exception {
        String account = URLDecoder.decode(session.getQueryString(), "UTF-8");
        account = account.replace("account=", "");
        if (StringUtils.isEmpty(account)) {
            LOGGER.error("\n 链接失败- {}", "客服号码为空");
            session.close();
        }
        if (redisCache == null) {
            redisCache = SpringUtils.getBean(RedisCache.class);
        }
//
        if (chatLogMapper == null) {
            chatLogMapper = SpringUtils.getBean(ChatLogMapper.class);
        }

        boolean semaphoreFlag = false;
        // 尝试获取信号量
        semaphoreFlag = SemaphoreUtils.tryAcquire(socketSemaphore);
        if (!semaphoreFlag) {
            // 未获取到信号量
            LOGGER.error("\n 当前在线人数超过限制数- {}", socketMaxOnlineCount);
            WebSocketUsers.sendMessageToUserByText(session, "当前在线人数超过限制数：" + socketMaxOnlineCount);
            session.close();
        } else {
            // 添加用户
            WebSocketUsers.put(session.getId(), session);
            LOGGER.info("\n 建立连接 - {}", account);
            String mes = JSON.toJSONString(new HashMap<String, String>() {{
                put("type", "pong");
                put("message", "ack");
            }});
//            session.getBasicRemote().sendText(JSON.toJSONString(new HashMap<String, String>() {{
//                put("type", "pong");
//                put("message", "ack");
//            }}));
            WebSocketUsers.sendMessageToUserByText(session, mes);
            // 使用try-catch包装Redis操作
            try {
                sessionMap.put(account, session);
                registerSession(account);
                LOGGER.info("WebSocket连接建立成功 - Session key: {}", account);
            } catch (Exception e) {
                LOGGER.error("sessionMap存储Session失败", e);
            }
        }
    }

    /**
     * 注册新的WebSocket连接
     */
    public void registerSession(String account) {
        if (account.contains("_")) {
            String[] numbers = account.split("_");
            if (numbers.length == 2) {
                String customerNumber = numbers[0];
                // 更新number到sessionKey的映射
                numberToSessionKeys.computeIfAbsent(customerNumber, k -> new ConcurrentHashSet<>())
                        .add(account);
            }
        }
    }


    /**
     * 获取指定number的所有有效Session
     */
    public List<Session> getSessionsByNumber(String number) {
        List<Session> sessions = new ArrayList<>();
        Set<String> sessionKeys = numberToSessionKeys.get(number);

        if (sessionKeys != null) {
            for (String sessionKey : sessionKeys) {
                Session session = sessionMap.get(sessionKey);
                if (session != null && session.isOpen()) {
                    sessions.add(session);
                } else {
                    // 清理无效的会话
                    sessionMap.remove(sessionKey);
                    sessionKeys.remove(sessionKey);
                }
            }
        }

        return sessions;
    }

    /**
     * 连接关闭时处理
     */
    @OnClose
    public void onClose(Session session) {
        LOGGER.info("\n 关闭连接 - {}", session);
        // 移除用户
        boolean removeFlag = WebSocketUsers.remove(session.getId());
        if (!removeFlag) {
            // 获取到信号量则需释放
            SemaphoreUtils.release(socketSemaphore);
        }

    }

    /**
     * 抛出异常时处理
     */
    @OnError
    public void onError(Session session, Throwable exception) throws Exception {
        if (session.isOpen()) {
            // 关闭连接
            session.close();
        }
        String sessionId = session.getId();
        LOGGER.info("\n 连接异常 - {}", sessionId);
//        LOGGER.info("\n 异常信息 - {}", exception);
        // 移出用户
        WebSocketUsers.remove(sessionId);
        // 获取到信号量则需释放
        SemaphoreUtils.release(socketSemaphore);
    }

    public void setMessage(String sessionKey, ReceivedMessage message) {
        try {
            Set<String> sessionKeySet = getRelatedAccountsByCustomerNumber(sessionKey);
            int successCount = 0;
            int failCount = 0;
            for (String relatedSessionKey : sessionKeySet) {
                Session session = sessionMap.get(relatedSessionKey);
                if (session != null && session.isOpen()) {
                    try {
                        String translation = WhatsAPPUtils.translateByTarget("zh-CN", message.getMessage());
                        if (StringUtils.isEmpty(translation))
                            return;
                        ChatLog chatLog = new ChatLog();
                        chatLog.setMsgId(message.getMsgId());
                        chatLog.setTranslateText(translation);
                        message.setTranslation(translation);
                        chatLogMapper.updateChatLogForTranslateTextBymsgId(chatLog);
                        successCount++;
                    } catch (Exception e) {
                        LOGGER.error("向账户 {} 发送消息失败", relatedSessionKey, e);
                        failCount++;
                    }
                    String mes = JSON.toJSONString(message);
                    WebSocketUsers.sendMessageToUserByText(session, mes);
                } else {
                    // 清理无效会话
                    cleanupInvalidSession(relatedSessionKey);
                    failCount++;
                }
            }
            LOGGER.info("消息发送完成: 成功 {}, 失败 {}", successCount, failCount);
        } catch (Exception e) {
            LOGGER.error("setMessage", e);
        }
    }

    /**
     * 清理无效的会话
     */
    private void cleanupInvalidSession(String sessionKey) {
        sessionMap.remove(sessionKey);
        String customerNumber = getCustomerNumberFromAccount(sessionKey);
        if (customerNumber != null) {
            Set<String> accounts = numberToSessionKeys.get(customerNumber);
            if (accounts != null) {
                accounts.remove(sessionKey);
                if (accounts.isEmpty()) {
                    numberToSessionKeys.remove(customerNumber);
                }
            }
        }
        LOGGER.debug("已清理无效会话: {}", sessionKey);
    }

    /**
     * 从account中提取customerNumber
     *
     * @param account 账户标识，格式为"customerNumber_xxx"
     * @return 提取的customerNumber，如果格式不匹配则返回null
     */
    public String getCustomerNumberFromAccount(String account) {
        if (account != null && account.contains("_")) {
            String[] parts = account.split("_");
            if (parts.length >= 2) {
                return parts[0]; // 返回customerNumber部分
            }
        }
        return null; // 格式不匹配返回null
    }

    /**
     * 根据已知的account获取相同customerNumber的所有accounts
     *
     * @param account 已知的账户标识
     * @return 相同customerNumber的所有账户集合，如果无法提取customerNumber则返回空集
     */
    public Set<String> getRelatedAccountsByAccount(String account) {
        // 从account中提取customerNumber
        String customerNumber = getCustomerNumberFromAccount(account);
        if (customerNumber == null) {
            return new HashSet<>(); // 如果无法提取customerNumber，返回空集
        }

        // 获取该customerNumber关联的所有accounts
        Set<String> relatedAccounts = numberToSessionKeys.get(customerNumber);
        if (relatedAccounts == null) {
            return new HashSet<>(); // 如果没有找到关联accounts，返回空集
        }

        return new HashSet<>(relatedAccounts); // 返回复制的集合，防止并发修改
    }

    /**
     * 根据已知的account获取相同customerNumber的所有accounts
     *
     * @param customerNumber 已知的账户标识
     * @return 相同customerNumber的所有账户集合，如果无法提取customerNumber则返回空集
     */
    public Set<String> getRelatedAccountsByCustomerNumber(String customerNumber) {
        // 获取该customerNumber关联的所有accounts
        Set<String> relatedAccounts = numberToSessionKeys.get(customerNumber);
        if (relatedAccounts == null) {
            return new HashSet<>(); // 如果没有找到关联accounts，返回空集
        }

        return new HashSet<>(relatedAccounts); // 返回复制的集合，防止并发修改
    }

    /**
     * 服务器接收到客户端消息时调用的方法
     */
    @OnMessage
    public void onMessage(String message, Session session) throws IOException {
        try {
            JSONObject jsonObject = JSON.parseObject(message);
            String type = jsonObject.getString("type");

            if ("ping".equals(type)) {
                try {
                    // 响应心跳
//                    session.getBasicRemote().sendText(JSON.toJSONString(Map.of("type", "pong", "message", "pong")
//                    ));
                    session.getBasicRemote().sendText(JSON.toJSONString(new HashMap<String, String>() {{
                        put("type", "pong");
                        put("message", "pong");
                    }}));
                    return;
                } catch (Exception e) {
                    LOGGER.error("发送pong消息失败: {}", e.getMessage());
                }
            }
            if ("pong".equals(type)) {
                // 响应心跳
//                lastPongTime.put(userId, System.currentTimeMillis());
                return;
            }

            LOGGER.info("收到消息：" + message);
            ReceivedMessage receivedMessage = objectMapper.readValue(message, ReceivedMessage.class);
            if (receivedMessage == null) {
                LOGGER.error("Deserialized ReceivedMessage is null");
                return;
            }

            if (receivedMessage.getFromNumber() == null) {
                LOGGER.error("FromNumber in ReceivedMessage is null");
                return;
            }
            String translation = WhatsAPPUtils.getTrans(receivedMessage.getToNumber(), receivedMessage.getMessage(), true);
            receivedMessage.setMessage(translation);
            CompletableFuture.runAsync(() -> {
                redisCache.setCacheObject("persona:fans:" + receivedMessage.getToNumber() + ":language", receivedMessage.getTargetLanguage());
                String url = waURLService.getContactNumberURL(receivedMessage.getFromNumber(), "apiSevicePath.secketSendmessage");
                LOGGER.info("发送地址：" + url);
                WhatsAPPUtils.sendMessage(url, receivedMessage.getFromNumber(), receivedMessage.getToNumber(), receivedMessage.getMessageType(),
                        receivedMessage.getMessage(), "1");
                clearMessage(receivedMessage.getFromNumber(), receivedMessage.getToNumber());

            });

            // 返回成功消息
//            session.getBasicRemote().sendText(JSON.toJSONString(Map.of("success", true, "type", "success", "message", message, "translation", translation)));
            Map<String, Object> responseMap = new HashMap<>();
            responseMap.put("success", true);
            responseMap.put("type", "success");
            responseMap.put("message", message);
            responseMap.put("translation", translation);
            session.getBasicRemote().sendText(JSON.toJSONString(responseMap));
        } catch (Exception e) {
            LOGGER.error("onMessage", e);
            ReceivedMessage receivedMessage = new ReceivedMessage();
            receivedMessage.setUserType(0);
            receivedMessage.setType("chat");
            receivedMessage.setMessage(null);
            String mes = JSON.toJSONString(receivedMessage);
            session.getBasicRemote().sendText(mes);
        }
    }


    //清除记录信息
    public void clearMessage(String customerNumber, String fansNumber) {
        redisCache.deleteObject("message:content:" + fansNumber + "-" + customerNumber + "customerNumber");
        redisCache.deleteObject("message:count:" + fansNumber + "-" + customerNumber + "customerNumber");
        redisCache.deleteObject("message:timestamp:" + fansNumber + "-" + customerNumber + "customerNumber");
    }

    @Scheduled(fixedRate = 30000) // 每30秒执行一次
    public void checkConnections() {
        String isOpen = redisCache.getCacheObject("Scheduled:open");
        if ("1".equals(isOpen)) {
            return;
        }
        for (Session session : WebSocketUsers.getUsers().values()) {
            if (!session.isOpen()) {
                LOGGER.warn("发现失效连接，userId: {}", getUserIdFromSession(session));
                try {
                    session.close();
                } catch (Exception e) {
                    LOGGER.error("关闭失效连接异常", e);
                }
            }
        }
    }

    private String getUserIdFromSession(Session session) {
        // 从session中获取userId的逻辑
        return session.getPathParameters().get("userId");
    }

}
